package com.cgy.utils.elasticsearch;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetRequest;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.IndicesClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHits;

import com.cgy.utils.base.arrays.ArraysHelper;
import com.cgy.utils.base.json.JsonHelper;
import com.cgy.utils.elasticsearch.action.CallBack;
import com.cgy.utils.elasticsearch.bulk.ElasticSearchBulkAdd;
import com.cgy.utils.elasticsearch.bulk.ElasticSearchBulkDelete;
import com.cgy.utils.elasticsearch.bulk.ElasticSearchBulkUpdate;
import com.cgy.utils.elasticsearch.condition.Condition;
import com.cgy.utils.elasticsearch.config.BulkConfig;
import com.cgy.utils.elasticsearch.config.IndexConfig;
import com.cgy.utils.elasticsearch.page.Page;
import com.cgy.utils.elasticsearch.search.ElasticSearchHelper;
import com.cgy.utils.elasticsearch.single.ElasticSearchAdd;
import com.cgy.utils.elasticsearch.single.ElasticSearchDelete;
import com.cgy.utils.elasticsearch.single.ElasticSearchGet;
import com.cgy.utils.elasticsearch.single.ElasticSearchSearch;
import com.cgy.utils.elasticsearch.single.ElasticSearchUpdate;
import com.cgy.utils.netx.config.ClientConfig;
import com.google.common.base.Joiner;

import lombok.extern.slf4j.Slf4j;

/**
 * @author CH
 */
@Slf4j
public class ElasticHelper<T> implements Cloneable {

    private static final String INDEX_KEY = "index";
    private static final String TYPE_KEY = "type";
    private static final String INDEX = "spider";
    private static final String TYPE = "doc";
    private static final String TIMESTAMP = "timestamp";

    private Class<? extends T> classes;
    private String className;

    private ClientConfig clientConfig = null;

    protected RestHighLevelClient highRestclient;
    protected RestClient lowRestClient;

    public ElasticHelper(String[] hosts) {
        ClientConfig clientConfig = new ClientConfig();
        clientConfig.setIp(Joiner.on(",").join(hosts));
        this.clientConfig = clientConfig;
    }

    public ElasticHelper(String hosts) {
        ClientConfig clientConfig = new ClientConfig();
        clientConfig.setIp(hosts);
        this.clientConfig = clientConfig;
    }

    public ElasticHelper(ClientConfig clientConfig) {
        this.clientConfig = clientConfig;
    }

    /**
     * @param classes
     * @return
     */
    public ElasticHelper classType(Class<? extends T> classes) {
        this.classes = classes;
        this.className = classes.getSimpleName();
        return this;
    }

    /**
     * @throws IOException
     */
    public void close() throws IOException {
        if (Objects.nonNull(highRestclient)) {
            highRestclient.close();
        }
    }

    /**
     * 初始化配置
     */
    public void configure() {
        HttpHost[] httpHosts = Arrays.stream(clientConfig.getIp().split(",")).map(host -> {
            return HttpHost.create(host);
        }).filter(Objects::nonNull).toArray(HttpHost[]::new);

        RestClientBuilder builder = RestClient.builder(httpHosts);
        
        builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                if (clientConfig.getConnectionTimeoutMs() > 0) {
                    requestConfigBuilder.setConnectTimeout(clientConfig.getConnectionTimeoutMs());
                }
                if (clientConfig.getConnectionRequestTimeout() > 0) {
                    requestConfigBuilder.setConnectionRequestTimeout(clientConfig.getConnectionRequestTimeout());
                }
                if (clientConfig.getSocketTimeout() > 0) {
                    requestConfigBuilder.setSocketTimeout(clientConfig.getSocketTimeout());
                }
                return requestConfigBuilder;
            }
        });

        builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                httpClientBuilder.setMaxConnTotal(clientConfig.getMaxConnTotal());
                httpClientBuilder.setMaxConnPerRoute(clientConfig.getMaxConnPerRoute());
                return httpClientBuilder;
            }
        });
        highRestclient = new RestHighLevelClient(builder);
        lowRestClient = highRestclient.getLowLevelClient();
    }

    
    /**
     * 创建索引
     *
     * @return
     */
    public boolean createIndex(IndexConfig indexConfig, CallBack<Boolean>... callBack) {
        String index = indexConfig.getIndex();
        if (checkIndexExists(index)) {
            log.error("index={}索引已经存在！", index);
            return false;
        }
        CreateIndexRequest request = new CreateIndexRequest(index);

        if (indexConfig.getShards() > 0 && indexConfig.getReplicas() > 0) {
            Settings.Builder builder = Settings.builder()
                    .put("index.number_of_shards", indexConfig.getShards())
                    .put("index.number_of_replicas", indexConfig.getReplicas());
            request.settings(builder);
        }
        try {
            if (StringUtils.isNotBlank(indexConfig.getType())) {
                request.mapping(indexConfig.getType(), XContentType.JSON);
            }
            if (!indexConfig.isAsync()) {
                CreateIndexResponse createIndexResponse = highRestclient.indices().create(request, RequestOptions.DEFAULT);
                // 指示是否所有节点都已确认请求
                boolean acknowledged = createIndexResponse.isAcknowledged();
                // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
                boolean shardsAcknowledged = createIndexResponse.isShardsAcknowledged();
                if (acknowledged || shardsAcknowledged) {
                    log.info("创建索引成功！索引名称为{}", createIndexResponse);
                    return true;
                }
            } else {
                highRestclient.indices().createAsync(request, RequestOptions.DEFAULT, new ActionListener<CreateIndexResponse>() {
                    @Override
                    public void onResponse(CreateIndexResponse createIndexResponse) {
                        // 指示是否所有节点都已确认请求
                        boolean acknowledged = createIndexResponse.isAcknowledged();
                        // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
                        boolean shardsAcknowledged = createIndexResponse.isShardsAcknowledged();
                        if (acknowledged || shardsAcknowledged) {
                            log.info("创建索引成功！索引名称为{}", createIndexResponse);
                        }
                        ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<Boolean>>() {
                            @Override
                            public void run(CallBack<Boolean> callBack) {
                                callBack.onResponse(acknowledged || shardsAcknowledged);
                            }
                        });
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.info("创建索引失败！原因:{}", e.getMessage());
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除索引
     *
     * @param indexConfig
     */
    public void deleteIndex(IndexConfig indexConfig, CallBack<Boolean>... callBack) {
        String index = indexConfig.getIndex();
        if (!checkIndexExists(index)) {
            log.error("index={}索引不存在！", index);
            return;
        }

        IndicesClient indices = highRestclient.indices();
        try {

            if (!indexConfig.isAsync()) {
                AcknowledgedResponse response = indices.delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT);
                if (response.isAcknowledged()) {
                    log.info("{} 索引删除成功！", index);
                }
            } else {
                indices.deleteAsync(new DeleteIndexRequest(index), RequestOptions.DEFAULT, new ActionListener<AcknowledgedResponse>() {
                    @Override
                    public void onResponse(AcknowledgedResponse response) {
                        boolean acknowledged = response.isAcknowledged();
                        if (acknowledged) {
                            log.info("{} 索引删除成功！", index);
                        }
                        ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<Boolean>>() {
                            @Override
                            public void run(CallBack<Boolean> callBack) {
                                callBack.onResponse(acknowledged);
                            }
                        });
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("删除失败！");
                    }
                });
            }
        } catch (ElasticsearchException ex) {
            if (ex.status() == RestStatus.NOT_FOUND) {
                log.error("{} 索引名不存在", index);
            }
            log.error("删除失败！");
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    /**
     * 添加文档
     *
     * @param indexConfig
     * @return
     */
    public boolean addDocument(IndexConfig indexConfig, CallBack<Boolean>... callBack) {
        String indexName = indexConfig.getIndex();
        String doc = indexConfig.getDoc();
        if (StringUtils.isBlank(doc)) {
            log.error("数据不存在. doc: {}", doc);
            return false;
        }
        if (!checkIndexExists(indexName)) {
            createIndex(indexConfig);
        }
        IndexRequest request = new IndexRequest();
        request.index(indexName);

        ArraysHelper.hasLength(indexConfig.getId(), new ArraysHelper.CallBack<String>() {
            @Override
            public void run(String s) {
                request.id(s);
            }
        });

        // request的opType默认是INDEX(传入相同id会覆盖原document，CREATE则会将旧的删除)
        // request.opType(DocWriteRequest.OpType.CREATE)
        request.source(doc, XContentType.JSON);

        if (!indexConfig.isAsync()) {
            IndexResponse response = null;
            try {
                response = highRestclient.index(request, RequestOptions.DEFAULT);
                return ElasticSearchAdd.response(response);
            } catch (ElasticsearchException e) {
                if (e.status() == RestStatus.CONFLICT) {
                    log.error("版本异常！");
                }
                log.error("\r\n index: {}, 文档新增失败！\r\n cause: {}", e.getIndex(), e.getDetailedMessage());
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            highRestclient.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
                @Override
                public void onResponse(IndexResponse response) {
                    boolean response1 = ElasticSearchAdd.response(response);
                    ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<Boolean>>() {
                        @Override
                        public void run(CallBack<Boolean> callBack) {
                            callBack.onResponse(response1);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {

                }
            });
        }
        return false;
    }

    /**
     * 获取文本
     *
     * @param indexConfig
     * @return
     */
    public String getDocument(IndexConfig indexConfig, CallBack<String>... callBack) {
        String index = indexConfig.getIndex();
        if (checkIndexExists(index)) {

            String id = indexConfig.getId();
            boolean realtime = indexConfig.isRealtime();
            boolean refresh = indexConfig.isRefresh();
            boolean async = indexConfig.isAsync();

            GetRequest request = new GetRequest();
            request.index(index);
            if (StringUtils.isNotBlank(id)) {
                request.id(id);
            }
            request.realtime(realtime).refresh(refresh);
            if (!async) {
                GetResponse response = null;
                try {
                    response = highRestclient.get(request, RequestOptions.DEFAULT);
                } catch (ElasticsearchException e) {
                    if (e.status() == RestStatus.NOT_FOUND) {
                        log.error("文档未找到，请检查参数！");
                    }
                    if (e.status() == RestStatus.CONFLICT) {
                        log.error("版本冲突！");
                    }
                    log.error("查找失败！");
                } catch (IOException e) {
                    e.printStackTrace();
                }

                ElasticSearchGet.response(response);
            } else {
                highRestclient.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                    @Override
                    public void onResponse(GetResponse response) {
                        String response1 = ElasticSearchGet.response(response);
                        ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<String>>() {
                            @Override
                            public void run(CallBack<String> stringCallBack) {
                                stringCallBack.onResponse(response1);
                            }
                        });
                    }

                    @Override
                    public void onFailure(Exception e) {
                    }
                });
            }
        }
        return null;
    }

    /**
     * 删除文档
     *
     * @param indexConfig
     * @return
     */
    public boolean deleteDocument(IndexConfig indexConfig, CallBack<Boolean>... callBack) {
        String index = indexConfig.getIndex();
        if (checkIndexExists(index)) {
            String id = indexConfig.getId();
            long version = indexConfig.getVersion();
            boolean async = indexConfig.isAsync();

            DeleteRequest request = new DeleteRequest();
            request.index(index);
            if (StringUtils.isNotBlank(id)) {
                request.id(id);
            }
            if (version > 0L) {
                request.version(version);
            }

            if (!async) {
                try {
                    DeleteResponse response = highRestclient.delete(request, RequestOptions.DEFAULT);
                    return ElasticSearchDelete.response(response);
                } catch (IOException e) {
                    e.printStackTrace();
                    return false;
                } catch (ElasticsearchException e) {
                    if (e.status() == RestStatus.CONFLICT) {
                        log.error("版本冲突！");
                    }
                    log.error("删除失败!");
                    return false;
                }
            } else {
                highRestclient.deleteAsync(request, RequestOptions.DEFAULT, new ActionListener<DeleteResponse>() {
                    @Override
                    public void onResponse(DeleteResponse deleteResponse) {
                        boolean response = ElasticSearchDelete.response(deleteResponse);
                        ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<Boolean>>() {
                            @Override
                            public void run(CallBack<Boolean> callBack) {
                                callBack.onResponse(response);
                            }
                        });
                    }

                    @Override
                    public void onFailure(Exception e) {
                    }
                });
            }
        }
        log.warn("index={} is not exist", index);
        return true;
    }

    /**
     * 通过一个JSON字符串更新文档(如果该文档不存在，则根据参数创建这个文档)
     *
     * @param indexConfig
     */
    public boolean updateDocument(IndexConfig indexConfig, CallBack<Boolean>... callBack) {
        String index = indexConfig.getIndex();
        String id = indexConfig.getId();
        if (StringUtils.isBlank(id)) {
            log.error("id must be exist");
            return false;
        }
        if (!checkIndexExists(index)) {
            createIndex(indexConfig);
        }

        UpdateRequest request = new UpdateRequest();
        request.index(index);
        request.id(id);
        request.doc(indexConfig.getDoc());
        // 如果要更新的文档不存在，则根据传入的参数新建一个文档
        request.docAsUpsert(true);

        if (indexConfig.isAsync()) {
            try {
                UpdateResponse response = highRestclient.update(request, RequestOptions.DEFAULT);
                return ElasticSearchUpdate.response(response);
            } catch (ElasticsearchException e) {
                if (e.status() == RestStatus.NOT_FOUND) {
                    log.error("不存在这个文档，请检查参数！");
                } else if (e.status() == RestStatus.CONFLICT) {
                    log.error("版本冲突异常！");
                }
                log.error("更新失败！");
                return false;
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            highRestclient.updateAsync(request, RequestOptions.DEFAULT, new ActionListener<UpdateResponse>() {
                @Override
                public void onResponse(UpdateResponse updateResponse) {
                    boolean response = ElasticSearchUpdate.response(updateResponse);
                    ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<Boolean>>() {
                        @Override
                        public void run(CallBack<Boolean> booleanCallBack) {
                            booleanCallBack.onResponse(response);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {

                }
            });
        }
        return false;
    }

    /**
     * 批量增加文档
     *
     * @return
     */
    public List<BulkItemResponse.Failure> bulkAddDocument(BulkConfig<List<BulkItemResponse.Failure>> bulkConfig, List<IndexConfig> indexConfig) throws IOException {
        if (null != indexConfig) {
            IndexConfig[] indexConfigs = new IndexConfig[indexConfig.size()];
            for (int i = 0; i < indexConfigs.length; i++) {
                indexConfigs[i] = indexConfig.get(i);
            }
            return bulkAddDocument(bulkConfig, indexConfigs);
        }
        return null;
    }

    /**
     * 批量增加文档
     *
     * @return
     */
    public List<BulkItemResponse.Failure> bulkAddDocument(BulkConfig<List<BulkItemResponse.Failure>> bulkConfig, IndexConfig... indexConfig) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        if (null != indexConfig && indexConfig.length > 0) {
            for (IndexConfig config : indexConfig) {
                String index = config.getIndex();
                String id = config.getId();
                if (StringUtils.isNotBlank(id)) {
                    IndexRequest indexRequest = new IndexRequest();
                    indexRequest.index(index);
                    indexRequest.id(id);
                    /*
                     * if(config.isOverride()) { indexRequest.opType(OpType.CREATE); }
                     */
                    indexRequest.source(config.getDoc(), XContentType.JSON);
                    bulkRequest.add(indexRequest);
                }
            }
        }
        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(bulkConfig.getSessionTimeout()));
        // 刷新策略
        bulkRequest.setRefreshPolicy(bulkConfig.getRefreshPolicy());

        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量增加操作失败！");
            return null;
        }

        if (!bulkConfig.isAsync()) {

            BulkResponse bulkResponse = highRestclient.bulk(bulkRequest, RequestOptions.DEFAULT);
            return ElasticSearchBulkAdd.response(bulkResponse);
        } else {
            highRestclient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
                @Override
                public void onResponse(BulkResponse bulkResponse) {
                    List<BulkItemResponse.Failure> response = ElasticSearchBulkAdd.response(bulkResponse);
                    ArraysHelper.hasLength(bulkConfig.getCallBack(), new ArraysHelper.CallBack<CallBack<List<BulkItemResponse.Failure>>>() {
                        @Override
                        public void run(CallBack<List<BulkItemResponse.Failure>> listCallBack) {
                            listCallBack.onResponse(response);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {
                }
            });
        }
        return null;
    }

    /**
     * 批量更新文档
     *
     * @return
     * @throws IOException
     */
    public List<BulkItemResponse.Failure> bulkUpdateDocument(BulkConfig<List<BulkItemResponse.Failure>> bulkConfig, List<IndexConfig> indexConfig) throws IOException {
        if (null != indexConfig) {
            IndexConfig[] indexConfigs = new IndexConfig[indexConfig.size()];
            for (int i = 0; i < indexConfigs.length; i++) {
                indexConfigs[i] = indexConfig.get(i);
            }
            return bulkUpdateDocument(bulkConfig, indexConfigs);
        }
        return null;
    }

    /**
     * 批量更新文档
     *
     * @return
     * @throws IOException
     */
    public List<BulkItemResponse.Failure> bulkUpdateDocument(BulkConfig<List<BulkItemResponse.Failure>> bulkConfig, IndexConfig... indexConfig) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        if (null != indexConfig && indexConfig.length > 0) {
            for (IndexConfig config : indexConfig) {
                String index = config.getIndex();
                String id = config.getId();
                if (StringUtils.isNotBlank(id)) {
                    UpdateRequest request = new UpdateRequest();
                    request.index(index);
                    request.id(id);
                    request.docAsUpsert(true);
                    request.doc(config.getDoc(), XContentType.JSON);
                    bulkRequest.add(request);
                }
            }
        }
        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(bulkConfig.getSessionTimeout()));
        // 刷新策略
        bulkRequest.setRefreshPolicy(bulkConfig.getRefreshPolicy());

        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量更新操作失败！");
            return null;
        }

        if (!bulkConfig.isAsync()) {

            BulkResponse bulkResponse = highRestclient.bulk(bulkRequest, RequestOptions.DEFAULT);
            return ElasticSearchBulkUpdate.response(bulkResponse);
        } else {
            highRestclient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
                @Override
                public void onResponse(BulkResponse bulkResponse) {
                    List<BulkItemResponse.Failure> response = ElasticSearchBulkUpdate.response(bulkResponse);
                    ArraysHelper.hasLength(bulkConfig.getCallBack(), new ArraysHelper.CallBack<CallBack<List<BulkItemResponse.Failure>>>() {
                        @Override
                        public void run(CallBack<List<BulkItemResponse.Failure>> listCallBack) {
                            listCallBack.onResponse(response);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {
                }
            });
        }
        return null;
    }
    /**
     * 批量删除文档
     *
     * @return
     * @throws IOException
     */
    public List<BulkItemResponse.Failure> bulkDeleteDocument(BulkConfig<List<BulkItemResponse.Failure>> bulkConfig, List<IndexConfig> indexConfig) throws IOException {
    	 if (null != indexConfig) {
             IndexConfig[] indexConfigs = new IndexConfig[indexConfig.size()];
             for (int i = 0; i < indexConfigs.length; i++) {
                 indexConfigs[i] = indexConfig.get(i);
             }
             return bulkDeleteDocument(bulkConfig, indexConfigs);
         }
         return null;
    }

    /**
     * 批量删除文档
     *
     * @return
     * @throws IOException
     */
    public List<BulkItemResponse.Failure> bulkDeleteDocument(BulkConfig<List<BulkItemResponse.Failure>> bulkConfig, IndexConfig... indexConfig) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        if (null != indexConfig && indexConfig.length > 0) {
            for (IndexConfig config : indexConfig) {
                String index = config.getIndex();
                String id = config.getId();
                if (StringUtils.isNotBlank(id)) {
                    DeleteRequest request = new DeleteRequest();
                    request.index(index);
                    request.id(id);
                    bulkRequest.add(request);
                }
            }
        }

        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(bulkConfig.getSessionTimeout()));
        // 刷新策略
        bulkRequest.setRefreshPolicy(bulkConfig.getRefreshPolicy());

        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量删除操作失败！");
            return null;
        }

        if (!bulkConfig.isAsync()) {
            BulkResponse bulkResponse = highRestclient.bulk(bulkRequest, RequestOptions.DEFAULT);
            return ElasticSearchBulkDelete.response(bulkResponse);
        } else {
            highRestclient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
                @Override
                public void onResponse(BulkResponse bulkItemResponses) {
                    List<BulkItemResponse.Failure> response = ElasticSearchBulkDelete.response(bulkItemResponses);
                    ArraysHelper.hasLength(bulkConfig.getCallBack(), new ArraysHelper.CallBack<CallBack<List<BulkItemResponse.Failure>>>() {
                        @Override
                        public void run(CallBack<List<BulkItemResponse.Failure>> listCallBack) {
                            listCallBack.onResponse(response);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {
                }
            });
        }

        return null;
    }

    /**
     * 批量删除文档
     *
     * @return
     * @throws IOException
     */
    public String bulkQueryDocument(BulkConfig<String> bulkConfig, IndexConfig... indexConfig) throws IOException {
        MultiGetRequest request = new MultiGetRequest();
        if (null != indexConfig && indexConfig.length > 0) {
            for (IndexConfig config : indexConfig) {
                String index = config.getIndex();
                String id = config.getId();
                if (StringUtils.isNotBlank(id)) {
                    request.add(new MultiGetRequest.Item(index, id));
                }
            }
        }

        request.realtime(bulkConfig.isRealtime()).refresh(bulkConfig.isRefresh());

        if (!bulkConfig.isAsync()) {
            MultiGetResponse response = highRestclient.mget(request, RequestOptions.DEFAULT);
            return JsonHelper.toJson(parseMGetResponse(response));
        } else {
            highRestclient.mgetAsync(request, RequestOptions.DEFAULT, new ActionListener<MultiGetResponse>() {
                @Override
                public void onResponse(MultiGetResponse response) {
                    List<Map<String, Object>> mapList = parseMGetResponse(response);
                    ArraysHelper.hasLength(bulkConfig.getCallBack(), new ArraysHelper.CallBack<CallBack<String>>() {
                        @Override
                        public void run(CallBack<String> stringCallBack) {
                            stringCallBack.onResponse(JsonHelper.toJson(mapList));
                        }
                    });

                }

                @Override
                public void onFailure(Exception e) {
                }
            });
        }
        return null;
    }

  
    /**
     * 检索数据
     *
     * @param condition
     * @return
     */
    public List<Map<String, Object>> queryAllByConditions(Condition condition, CallBack<List<Map<String, Object>>>... callBack) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        if (StringUtils.isAllBlank(condition.getLevel(), condition.getMessage())) {
            log.error("level(日志级别)和messageKey(日志信息关键字)不能同时为空！");
            return resultList;
        }

        SearchRequest request = ElasticSearchHelper.search(condition);
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(condition.getScrollTime()));

        if (!condition.isAsync()) {
            try {
                SearchResponse response = highRestclient.search(request, RequestOptions.DEFAULT);
                String scroll1 = ElasticSearchSearch.scroll(resultList, response, scroll, highRestclient);
                // 清理 scroll 上下文
                ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                clearScrollRequest.addScrollId(scroll1);
                highRestclient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return resultList;
        } else {
            highRestclient.searchAsync(request, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<List<Map<String, Object>>>>() {
                        @Override
                        public void run(CallBack<List<Map<String, Object>>> listCallBack) {
                            String scroll1 = ElasticSearchSearch.scroll(resultList, response, scroll, highRestclient);
                            // 清理 scroll 上下文
                            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                            clearScrollRequest.addScrollId(scroll1);
                            try {
                                highRestclient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }

                            listCallBack.onResponse(resultList);
                        }
                    });

                }

                @Override
                public void onFailure(Exception e) {

                }
            });
        }
        return resultList;
    }

    /**
     * 检索数据
     *
     * @param condition
     * @return
     */
    public Page<List<Map<String, Object>>> queryPage(Condition condition, CallBack<Page<List<Map<String, Object>>>>... callBack) {
        Page<List<Map<String, Object>>> listPage = new Page();
        if (StringUtils.isAllBlank(condition.getLevel(), condition.getMessage())) {
            log.error("参数level(日志级别)和messageKey(日志信息关键字)不能同时为空！");
            return listPage;
        }
        SearchRequest search = ElasticSearchHelper.search(condition);
        if (!condition.isAsync()) {
            try {
                SearchResponse response = highRestclient.search(search, RequestOptions.DEFAULT);
                SearchHits hits = response.getHits();
                int total = (int) hits.getTotalHits().value;
                List<Map<String, Object>> mapList = ElasticSearchSearch.queryList(response);
                listPage.setTotal(total);
                listPage.setPageNum(condition.getStart());
                listPage.setPageSize(condition.getSize());
                listPage.setResult(mapList);
                return listPage;
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            highRestclient.searchAsync(search, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<Page<List<Map<String, Object>>>>>() {
                        @Override
                        public void run(CallBack<Page<List<Map<String, Object>>>> pageCallBack) {
                            SearchHits hits = response.getHits();
                            int total = (int) hits.getTotalHits().value;
                            List<Map<String, Object>> mapList = ElasticSearchSearch.queryList(response);
                            listPage.setTotal(total);
                            listPage.setPageNum(condition.getStart());
                            listPage.setPageSize(condition.getSize());
                            listPage.setResult(mapList);
                            pageCallBack.onResponse(listPage);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {

                }
            });
        }
        return listPage;
    }

    /**
     * 检索数据
     *
     * @param condition
     * @return
     */
    public List<Map<String, Object>> queryList(Condition condition, CallBack<List<Map<String, Object>>>... callBack) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        /*
         * if (StringUtils.isAllBlank(condition.getLevel(), condition.getMessage())) {
         * log.error("参数level(日志级别)和messageKey(日志信息关键字)不能同时为空！"); return resultList; }
         */
        SearchRequest search = ElasticSearchHelper.search(condition);
        log.info(search.source().toString());
        if (!condition.isAsync()) {
            try {
                SearchResponse search1 = highRestclient.search(search, RequestOptions.DEFAULT);
                return ElasticSearchSearch.queryList(search1);
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            highRestclient.searchAsync(search, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    ArraysHelper.hasLengthAndGetFirst(callBack, new ArraysHelper.CallBack<CallBack<List<Map<String, Object>>>>() {
                        @Override
                        public void run(CallBack<List<Map<String, Object>>> listCallBack) {
                            List<Map<String, Object>> mapList = ElasticSearchSearch.queryList(response);
                            listCallBack.onResponse(mapList);
                        }
                    });
                }

                @Override
                public void onFailure(Exception e) {

                }
            });
        }
        return resultList;
    }

    /**
     * @param response
     * @return
     */
    private List<Map<String, Object>> parseMGetResponse(MultiGetResponse response) {
        List<Map<String, Object>> list = new ArrayList<>();
        MultiGetItemResponse[] responses = response.getResponses();
        for (MultiGetItemResponse item : responses) {
            GetResponse getResponse = item.getResponse();
            if (Objects.nonNull(getResponse)) {
                if (!getResponse.isExists()) {
                    log.error("\"index={}, type={}, id={}\"的文档查找失败，请检查参数！", getResponse.getIndex(), getResponse.getType(), getResponse.getId());
                } else {
                    list.add(getResponse.getSourceAsMap());
                }
            } else {
                MultiGetResponse.Failure failure = item.getFailure();
                ElasticsearchException e = (ElasticsearchException) failure.getFailure();
                if (e.status() == RestStatus.NOT_FOUND) {
                    log.error("\"index={}, type={}, id={}\"的文档不存在！", failure.getIndex(), failure.getType(), failure.getId());
                } else if (e.status() == RestStatus.CONFLICT) {
                    log.error("\"index={}, type={}, id={}\"的文档版本冲突！", failure.getIndex(), failure.getType(), failure.getId());
                }
            }
        }
        return list;
    }

    /**
     * 设置文档的静态映射(主要是为 message字段 设置ik分词器)
     *
     * @param indexConfig
     * @throws IOException
     */
    public void setFieldsMapping(IndexConfig indexConfig) {
        String index = indexConfig.getIndex();
        String type = indexConfig.getType();
        PutMappingRequest request = new PutMappingRequest(index).type(type);
        try {
            request.source(generateBuilder());
            AcknowledgedResponse acknowledgedResponse = highRestclient.indices().putMapping(request, RequestOptions.DEFAULT);
            if (acknowledgedResponse.isAcknowledged()) {
                log.info("已成功对\"index={}, type={}\"的文档设置类型映射！", index, type);
            }
        } catch (IOException e) {
            log.error("\"index={}, type={}\"的文档设置类型映射失败，请检查参数！", index, type);
        }
    }


    /**
     * 开启索引
     *
     * @param indexConfig
     * @throws IOException
     */
    public void openIndex(IndexConfig indexConfig) throws IOException {
        String index = indexConfig.getIndex();
        if (!checkIndexExists(index)) {
            log.error("索引不存在！");
            return;
        }
        OpenIndexRequest request = new OpenIndexRequest(index);
        OpenIndexResponse response = highRestclient.indices().open(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            log.info("{} 索引开启成功！", index);
        }
    }

    /**
     * 关闭索引
     *
     * @param indexConfig
     * @throws IOException
     */
    public void closeIndex(IndexConfig indexConfig) throws IOException {
        String index = indexConfig.getIndex();
        if (!checkIndexExists(index)) {
            log.error("索引不存在！");
            return;
        }
        CloseIndexRequest request = new CloseIndexRequest(index);
        AcknowledgedResponse response = highRestclient.indices().close(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged()) {
            log.info("{} 索引已关闭！", index);
        }
    }

    /**
     * 判断索引是否存在
     *
     * @param indexs
     * @return
     * @throws IOException
     */
    public boolean checkIndexExists(String... indexs) {
        GetIndexRequest request = new GetIndexRequest(indexs);
        try {
            return highRestclient.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("判断索引是否存在，操作异常！");
        }
        return false;
    }

    private XContentBuilder generateBuilder() throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        builder.startObject("properties");
        builder.startObject("message");
        builder.field("type", "text");
        // 为message字段，设置分词器为 ik_smart(最粗粒度)
        builder.field("analyzer", "ik_smart");
        builder.endObject();
        builder.startObject(TIMESTAMP);
        builder.field("type", "date");
        // 设置 日志时间的格式为  毫秒数的long类型
        builder.field("format", "epoch_millis");
        builder.endObject();
        builder.endObject();
        builder.endObject();
        return builder;
    }

}
